package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: guoml
 * @Description: 针对企查查t_ipo_company_income_statement表和聚源LC_IncomeStatementAll，LC_STIBIncomeState表进行数据映射开发
 * @Date: 2021/12/7 12:51
 */

public class TIpoCompanyIncomeStatement {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_ipo_company_income_statement").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        //获取中数增量数据表中的新增数据（left join）
        Dataset<Row> insert_data = spark.sql("SELECT md5(concat(100,a.ID)) as ID,\n" +
                "       b.SecuCode,\n" +
                "       a.TotalOperatingRevenue,\n" +
                "       a.OperatingRevenue,\n" +
                "       a.NetCommissionIncome,\n" +
                "       a.NetProxySecuIncome,\n" +
                "       a.InterestIncome,\n" +
                "       a.PremiumsEarned,\n" +
                "       a.CommissionIncome,\n" +
                "       a.PremiumsIncome,\n" +
                "       a.ReinsuranceIncome,\n" +
                "       a.Reinsurance,\n" +
                "       a.UnearnedPremiumReserve,\n" +
                "       '' as bank_intn_reve,\n" +
                "       '' as bank_int_reve,\n" +
                "       '' as bank_int_exp,\n" +
                "       a.NetCommissionIncome,\n" +
                "       a.CommissionIncome,\n" +
                "       a.CommissionExpense,\n" +
                "       a.NetSubIssueSecuIncome,\n" +
                "       a.NetTrustIncome,\n" +
                "       '' as fina_consult,\n" +
                "       '' as sponsor,\n" +
                "       a.NetFundMgtIncome,\n" +
                "       '' as fund_sale,\n" +
                "       a.NetProxySecuIncome,\n" +
                "       '' as commn_reveother,\n" +
                "       a.NetInterestIncome,\n" +
                "       a.InvestIncome,\n" +
                "       a.InvestIncomeAssociates,\n" +
                "       a.FairValueChangeIncome,\n" +
                "       a.OCIFairValue,\n" +
                "       a.ExchangeIncome,\n" +
                "       a.OtherOperatingRevenue,\n" +
                "       a.TotalOperatingCost,\n" +
                "       (case when a.OperatingCost is not null then a.OperatingCost else a.OperatingPayout end) as operate_exp,\n" +
                "       a.RefundedPremiums,\n" +
                "       a.NetClaimIncurred,\n" +
                "       a.NetPremiumReserve,\n" +
                "       a.CompensationExpense,\n" +
                "       a.AmortizationExpense,\n" +
                "       a.PremiumReserve,\n" +
                "       a.AmortizationPremiumReserve,\n" +
                "       a.PolicyDividendPayout,\n" +
                "       a.ReinsuranceCost,\n" +
                "       a.OperatingTaxSurcharges,\n" +
                "       a.OperatingExpense,\n" +
                "       a.AdministrationExpense,\n" +
                "       a.OperatingAndAdminExpense,\n" +
                "       a.AmortizationReinsuranceCost,\n" +
                "       a.InterestExpense,\n" +
                "       (case when a.InsuranceCommissionExpense is not null then a.InsuranceCommissionExpense else a.CommissionExpense end) as comm_exp,\n" +
                "       a.RAndD,\n" +
                "       a.FinancialExpense,\n" +
                "       a.OtherOperatingCost,\n" +
                "       a.AssetImpairmentLoss,\n" +
                "       a.OperatingProfit,\n" +
                "       a.NonoperatingIncome,\n" +
                "       '' as nonlasset_reve,\n" +
                "       a.NonoperatingExpense,\n" +
                "       a.NonCurrentAssetssDealLoss,\n" +
                "       a.TotalProfit,\n" +
                "       a.IncomeTaxCost,\n" +
                "       '' as combinednet_profitb,\n" +
                "       a.NetProfit,\n" +
                "       a.NPParentCompanyOwners,\n" +
                "       a.MinorityProfit,\n" +
                "       c.NetProfitCut,\n" +
                "       a.BasicEPS,\n" +
                "       a.DilutedEPS,\n" +
                "       a.OtherCompositeIncome,\n" +
                "       a.OCIParentCompanyOwners,\n" +
                "       a.OCIMinorityOwners,\n" +
                "       a.TotalCompositeIncome,\n" +
                "       a.CIParentCompanyOwners,\n" +
                "       a.CIMinorityOwners,\n" +
                "       '人民币' as currency,\n" +
                "       '' as security_code,\n" +
                "       '' as type,\n" +
                "       '' as created_date,\n" +
                "       '' as updated_date,\n" +
                "       a.EndDate,\n" +
                "       '' as report_encrypt_key,\n" +
                "       regexp_replace(current_date(),'-',''),\n" +
                "       \"1\" as isadd\n" +
                "FROM\n" +
                "  (select a.* from (select a.*,row_number ( ) over ( PARTITION BY CompanyCode,EndDate ORDER BY InfoPublDate desc,IfAdjusted) AS adjust from a_dc_ep_ods.LC_IncomeStatementAll a where IfMerged = 1 and IfComplete = 1 and IfAdjusted in (1,2) and InfoSourceCode in (110101,110102,110103,110104))a where a.adjust = 1)a\n" +
                "INNER JOIN\n" +
                "  (select * from a_dc_ep_ods.SecuMain where ListedState = 1 and ListedSector not in (3,7) and SecuCategory = 1) b on a.CompanyCode = b.CompanyCode\n" +
                "LEFT JOIN\n" +
                "  a_dc_ep_ods.LC_MainIndexNew c on a.CompanyCode = c.CompanyCode and a.EndDate = c.EndDate").union(spark.sql("SELECT md5(concat(300,a.ID)) as ID,\n" +
                "       b.SecuCode,\n" +
                "       a.TotalOperatingRevenue,\n" +
                "       a.OperatingRevenue,\n" +
                "       a.NetCommissionIncome,\n" +
                "       a.NetProxySecuIncome,\n" +
                "       a.InterestIncome,\n" +
                "       a.PremiumsEarned,\n" +
                "       a.CommissionIncome,\n" +
                "       a.PremiumsIncome,\n" +
                "       a.ReinsuranceIncome,\n" +
                "       a.Reinsurance,\n" +
                "       a.UneaPremRes as UnearnedPremiumReserve,\n" +
                "       '' as bank_intn_reve,\n" +
                "       '' as bank_int_reve,\n" +
                "       '' as bank_int_exp,\n" +
                "       a.NetCommissionIncome,\n" +
                "       a.CommissionIncome,\n" +
                "       a.CommissionExpense,\n" +
                "       a.NetSubIssueSecuIncome,\n" +
                "       a.NetTrustIncome,\n" +
                "       '' as fina_consult,\n" +
                "       '' as sponsor,\n" +
                "       a.NetFundMgtIncome,\n" +
                "       '' as fund_sale,\n" +
                "       a.NetProxySecuIncome,\n" +
                "       '' as commn_reveother,\n" +
                "       a.NetInterestIncome,\n" +
                "       a.InvInc as InvestIncome,\n" +
                "       a.InvestIncomeAssociates,\n" +
                "       a.FairValChaInc as FairValueChangeIncome,\n" +
                "       a.OCIFairValue,\n" +
                "       a.ExchangeIncome,\n" +
                "       a.OtherOperatingRevenue,\n" +
                "       a.TotalOperatingCost,\n" +
                "       a.OperatingCost as operate_exp,\n" +
                "       a.RefundedPremiums,\n" +
                "       a.NetClaimIncurred,\n" +
                "       a.NetPremiumReserve,\n" +
                "       a.CompensationExpense,\n" +
                "       a.AmortizationExpense,\n" +
                "       a.PremiumReserve,\n" +
                "       a.AmorPremRes as AmortizationPremiumReserve,\n" +
                "       a.PolicyDividendPayout,\n" +
                "       a.ReinsuranceCost,\n" +
                "       a.OperatingTaxSurcharges,\n" +
                "       a.OperatingExpense,\n" +
                "       a.AdministrationExpense,\n" +
                "       a.OpeAndAdmExp as OperatingAndAdminExpense,\n" +
                "       a.AmorReinCost as AmortizationReinsuranceCost,\n" +
                "       a.InterestExpense,\n" +
                "       a.InsComExpense as comm_exp,\n" +
                "       a.RAndD,\n" +
                "       a.FinancialExpense,\n" +
                "       a.OtherOperatingCost,\n" +
                "       a.AssetImpairmentLoss,\n" +
                "       a.OperatingProfit,\n" +
                "       a.NonoperatingIncome,\n" +
                "       '' as nonlasset_reve,\n" +
                "       a.NonoperatingExpense,\n" +
                "       a.NonCurAssDLoss as NonCurrentAssetssDealLoss,\n" +
                "       a.TotalProfit,\n" +
                "       a.IncomeTax as IncomeTaxCost,\n" +
                "       '' as combinednet_profitb,\n" +
                "       a.NetProfit,\n" +
                "       a.NPParentCompanyOwners,\n" +
                "       a.MinorityProfit,\n" +
                "       c.NetProfitCut,\n" +
                "       a.BasicEPS,\n" +
                "       a.DilutedEPS,\n" +
                "       a.OCIAfterTax as OtherCompositeIncome,\n" +
                "       a.OCIATPComOWNR as OCIParentCompanyOwners,\n" +
                "       a.OCIATMinorOwner as OCIMinorityOwners,\n" +
                "       a.TotalCompositeIncome,\n" +
                "       a.CIParentCompanyOwners,\n" +
                "       a.CIMinorityOwners,\n" +
                "       '人民币' as currency,\n" +
                "       '' as security_code,\n" +
                "       '' as type,\n" +
                "       '' as created_date,\n" +
                "       '' as updated_date,\n" +
                "       a.EndDate,\n" +
                "       '' as report_encrypt_key,\n" +
                "       regexp_replace(current_date(),'-',''),\n" +
                "       \"1\" as isadd\n" +
                "FROM\n" +
                "  (select a.* from (select a.*,row_number ( ) over ( PARTITION BY CompanyCode,EndDate ORDER BY InfoPublDate desc,IfAdjusted) AS adjust from a_dc_ep_ods.LC_STIBIncomeState a where IfMerged = 1 and IfComplete = 1 and IfAdjusted in (1,2) and InfoSourceCode in (110101,110102,110103,110104))a where a.adjust = 1)a\n" +
                "INNER JOIN\n" +
                "  (select * from a_dc_ep_ods.SecuMain where ListedState = 1 and ListedSector =7 and SecuCategory = 1) b on a.CompanyCode = b.CompanyCode\n" +
                "LEFT JOIN\n" +
                "  (select a.* from (select a.*,row_number ( ) over ( PARTITION BY CompanyCode,EndDate ORDER BY InfoPublDate desc,IfAdjusted) AS adjust from a_dc_ep_ods.LC_STIBMainIndex a where IfMerged = 1 and InfoSourceCode in (110101,110102,110103,110104))a where a.adjust = 1)c on a.CompanyCode = c.CompanyCode and a.EndDate = c.EndDate")).union(spark.sql("SELECT md5(concat(200,a.ID)) as ID,\n" +
                "       b.SecuCode,\n" +
                "       a.TotalOperateRev,\n" +
                "       a.OperatRevenue,\n" +
                "       a.NetComIncome,\n" +
                "       a.NetProxySecInc,\n" +
                "       a.InterestIncome,\n" +
                "       a.PremiumsEarned,\n" +
                "       a.ComIncome,\n" +
                "       a.PremiumsIncome,\n" +
                "       a.ReinsurIncome,\n" +
                "       a.Reinsurance,\n" +
                "       a.UPR as UnearnedPremiumReserve,\n" +
                "       '' as bank_intn_reve,\n" +
                "       '' as bank_int_reve,\n" +
                "       '' as bank_int_exp,\n" +
                "       a.NetComIncome,\n" +
                "       a.ComIncome,\n" +
                "       a.ComExpense,\n" +
                "       a.NetIssueSecInc,\n" +
                "       a.NetTrustIncome,\n" +
                "       '' as fina_consult,\n" +
                "       '' as sponsor,\n" +
                "       '' as fund_manage,\n" +
                "       '' as fund_sale,\n" +
                "       a.NetProxySecInc,\n" +
                "       '' as commn_reveother,\n" +
                "       a.NetIntIncome,\n" +
                "       a.InvestIncome,\n" +
                "       a.InvestIncAssoci,\n" +
                "       a.FVChangeInc as FairValueChangeIncome,\n" +
                "       a.OCIFairValue,\n" +
                "       a.ExchangeIncome,\n" +
                "       a.OtherOR,\n" +
                "       a.TotalOC,\n" +
                "       a.OperatingPayout as operate_exp,\n" +
                "       a.RefundedPremium,\n" +
                "       a.NetClaimIncurred,\n" +
                "       a.InsuranceContractReser,\n" +
                "       a.CompensationExs,\n" +
                "       a.AmortizationExs,\n" +
                "       a.PremiumReserve,\n" +
                "       a.AmortPremRSRV as AmortizationPremiumReserve,\n" +
                "       a.PolicyDividPay,\n" +
                "       a.ReinsuranceCost,\n" +
                "       a.OperateTax,\n" +
                "       a.OperatingExs,\n" +
                "       a.AdminExs,\n" +
                "       a.OperateAdminExs as OperatingAndAdminExpense,\n" +
                "       a.AmortReinsCost as AmortizationReinsuranceCost,\n" +
                "       a.InterestExpense,\n" +
                "       a.InsurCommExs as comm_exp,\n" +
                "       a.DevelopmentExpense,\n" +
                "       a.FinancialExs,\n" +
                "       a.OtherOC,\n" +
                "       a.Assetdevalue,\n" +
                "       a.OperatingProfit,\n" +
                "       a.NonoperateInc,\n" +
                "       a.NonCurAsDEarn as nonlasset_reve,\n" +
                "       a.NonoperateExs,\n" +
                "       a.NCurAssetDeLoss as NonCurrentAssetssDealLoss,\n" +
                "       a.TotalProfit,\n" +
                "       a.IncomeTaxCost,\n" +
                "       a.NProfitByMergedBefMerg as combinednet_profitb,\n" +
                "       a.NetProfit,\n" +
                "       a.NPParComOwners,\n" +
                "       a.MinorityProfit,\n" +
                "       c.NetProfitCut,\n" +
                "       a.BasicEPS,\n" +
                "       a.DilutedEPS,\n" +
                "       a.OCIAfterTax as OtherCompositeIncome,\n" +
                "       a.OCIATParComOWNR as OCIParentCompanyOwners,\n" +
                "       a.OCIATMinorOwner as OCIMinorityOwners,\n" +
                "       a.TotalCI,\n" +
                "       a.CIParComOwners,\n" +
                "       a.CIMinorityOwner,\n" +
                "       '人民币' as currency,\n" +
                "       '' as security_code,\n" +
                "       '' as type,\n" +
                "       '' as created_date,\n" +
                "       '' as updated_date,\n" +
                "       a.EndDate,\n" +
                "       '' as report_encrypt_key,\n" +
                "       regexp_replace(current_date(),'-',''),\n" +
                "       \"1\" as isadd\n" +
                "FROM\n" +
                "  (select a.* from (select a.*,row_number ( ) over ( PARTITION BY CompanyCode,EndDate ORDER BY InfoPublDate desc,IfAdjusted) AS adjust from a_dc_ep_ods.NQ_IncStatement a where IfMerged = 1 and IfComplete = 1 and IfAdjusted in (1,2) and InfoSource in (2,3,4,5))a where a.adjust = 1)a\n" +
                "INNER JOIN\n" +
                "  (select * from a_dc_ep_ods.SecuMain where ListedState = 1 and ListedSector =3 and SecuCategory = 1) b on a.CompanyCode = b.CompanyCode\n" +
                "LEFT JOIN\n" +
                "  (select a.* from (select a.*,row_number ( ) over ( PARTITION BY CompanyCode,EndDate ORDER BY Mark) AS rn from a_dc_ep_ods.NQ_MainIndexNew a)a where a.rn = 1)c on a.CompanyCode = c.CompanyCode and a.EndDate = c.EndDate"));
        insert_data.createOrReplaceTempView("tmp_t_ipo_company_income_statement1");
        spark.sql("insert overwrite table a_dc_ep_incr.t_ipo_company_income_statement select * from tmp_t_ipo_company_income_statement1");
        spark.stop();
    }
}
